package com.guandata.cdc.model.source;

import com.guandata.cdc.helper.ParamsHelper;
import lombok.ToString;
import org.apache.flink.api.java.utils.ParameterTool;

import java.util.List;
import java.util.UUID;

@ToString
public abstract class AbstractSourceConfig {
    private static final String BIZ_DATA_SOURCE = "biz-data-source";

    private static final String DEFAULT_CATALOG_NAME = "%s_%s_catalog_%s";

    public final String sourceHostname;

    public final String sourceUsername;

    public final int sourcePort;

    public final String sourcePassword;

    public final String sourceDatabase;

    public final String sourceName;

    public final String rawSourceTable;

    public final List<String> dropColumns;

    protected AbstractSourceConfig(ParameterTool parameterTool) {
        this.sourceHostname = parameterTool.getRequired("source.hostname");
        this.sourceUsername = parameterTool.getRequired("source.username");
        this.sourcePort = parameterTool.getInt("source.port");
        this.sourcePassword = parameterTool.getRequired("source.password");
        this.sourceDatabase = parameterTool.getRequired("source.database");
        this.sourceName = ParamsHelper.getInnerName(parameterTool.get("source.name", BIZ_DATA_SOURCE));
        this.rawSourceTable = parameterTool.getRequired("source.table");
        this.dropColumns = ParamsHelper.getListFromString(parameterTool.get("source.drop-columns"));
    }

    public String getCatalogName(String title) {
        return String.format(DEFAULT_CATALOG_NAME, title, sourceDatabase, UUID.randomUUID());
    }
}
